/*************************************************************************************
* 	 Copyright (C) 2010 by Information Systems Group, Saarland University  			*
*    http://infosys.cs.uni-saarland.de												*
* 	 																				*
* 	 This file is part of Hadoop++.												 	*
*																					*
*    Hadoop++ is free software: you can redistribute it and/or modify				*
*    it under the terms of the GNU Lesser General Public License as published by	*
*    the Free Software Foundation, either version 3 of the License, or				*
*    (at your option) any later version.											*
*																					*
*    Hadoop++ is distributed in the hope that it will be useful,					*
*    but WITHOUT ANY WARRANTY; without even the implied warranty of					*
*    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the					*
*    GNU Lesser General Public License for more details.							*
*																					*
*    You should have received a copy of the GNU Lesser General Public License		*
*    along with Hadoop++.  If not, see <http://www.gnu.org/licenses/>.				*
*************************************************************************************/
package unisb.cs.core.index.structure;

import unisb.cs.core.binary.utils.BinaryUtils;
import unisb.cs.data.types.DataTypes;

/*******************************************************************************
 * This class is based on the CSS-Trees originally implemented in C. See
 * copyright notice below.
 * <p>
 * Cache-Sensitive B+-Tree code.
 * <p>
 * Written by Jun Rao (junr@cs.columbia.edu)
 * <p>
 * Sep. 25, 1999
 * <p>
 * This code is copyrighted by the Trustees of Columbia University in the City
 * of New York.
 * <p>
 * Permission to use and modify this code for noncommercial purposes is granted
 * provided these headers are included unchanged.
 ******************************************************************************/

/**
 * A CSS tree.
 * <p>
 * key type is int, value type is int.
 */
public class CSSTreeWithLongOffsets {

	/** the value that is returned when the key is not found. */
	public static final int NOT_FOUND = Integer.MIN_VALUE;

	/** the array containing the directory. */
	private final byte[] internalNodes;

	/** index of the first node that is in the lowest level of the directory. */
	private final int nInternal;

	/**
	 * index of the first node where the key should be found in the first half
	 * of the data array.
	 */
	private final int thres;

	/** log2 of number of keys per internal node. */
	public final int mbits;

	/** number of keys per internal node. */
	private final int keyPNode;

	/** number of tuples per leaf. */
	private final int tuplesPLeaf;

	/** the number of elements this index contains. */
	private final int size;

	/** the array containing the data. */
	/*
	 * private final byte[] content;
	 */

	/** length of each tuple record */
	private final int recLen;

	/** data type of the index */
	private final int keyType;

	/** length of the key in bytes */
	private final int keySize;

	/** offset of the key within a record */
	private final int keyOffset;

	public final static int OFFSET_BYTES = 8;

	/**
	 * Constructor. Allows to set different internal node sizes.
	 * 
	 * @param content the array containing the data (will be captured!!!)
	 * @param mbits log2 of the number of keys per internal node
	 */
	public CSSTreeWithLongOffsets(final byte[] content, final int mbits, final int recLen,
			final int keyType, final int keySize, final int keyOffset) {
		this.recLen = recLen;
		this.keyType = keyType;
		this.keySize = keySize;
		this.keyOffset = keyOffset;

		this.size = content.length / recLen;

		this.mbits = mbits;
		keyPNode = 1 << mbits;
		tuplesPLeaf = keyPNode >> 1;

		int k, B, x, nonLeaf, mask, branch;
		int i, j, child, l, exp9k;
		// int[] b;
		byte[] b;
		mask = keyPNode - 1;
		branch = keyPNode + 1;
		B = (int) Math.ceil((double) size / (double) tuplesPLeaf);
		k = (int) Math.ceil(Math.log10((double) B)
				/ Math.log10((double) branch));
		exp9k = (int) Math.pow(branch, k);
		x = (exp9k - B) / keyPNode;

		nonLeaf = ((exp9k - 1) >> mbits) - x;
		nonLeaf = nonLeaf > 0 ? nonLeaf : 0;
		nInternal = nonLeaf;
		thres = ((exp9k - 1) >> mbits);

		b = new byte[(nonLeaf * keyPNode) * keySize];
		// System.out.println("size="+size+", nInternal="+nInternal);

		for (i = (nInternal) - 1; i >= 0; i--) {
			for (j = i * keyPNode + keyPNode - 1; j >= i * keyPNode; j--) {
				child = branch * i + (j & mask) + 1;
				while (child < nonLeaf)
					child = branch * child + branch;
				l = child - (thres); // l is the node index
				l = l * tuplesPLeaf; // l is the entry index
				if (l < 0) {
					l += size;
					copyIndexKeyAtPos(b, j * keySize, content,
							(l + tuplesPLeaf - 1) * recLen + keyOffset);
				} else {
					if (l <= size - (x + 1) * tuplesPLeaf) {
						final int temp = (l + tuplesPLeaf - 1);
						copyIndexKeyAtPos(b, j * keySize, content,
								(temp >= size ? size - 1 : temp) * recLen
										+ keyOffset);

					} else
						copyIndexKeyAtPos(b, j * keySize, content, (size - x
								* tuplesPLeaf - 1)
								* recLen + keyOffset);
				}
			}
		}
		this.internalNodes = b;
	}

	public CSSTreeWithLongOffsets(final byte[] index, final int contentSize, final int mbits,
			final int recLen, final int keyType, final int keySize,
			final int keyOffset) {
		this.recLen = recLen + OFFSET_BYTES;
		this.keyType = keyType;
		this.keySize = keySize + OFFSET_BYTES;
		this.keyOffset = keyOffset;
		this.internalNodes = index;
		this.size = contentSize;

		this.mbits = mbits;
		keyPNode = 1 << mbits;
		tuplesPLeaf = keyPNode >> 1;

		int k, B, x, nonLeaf, branch;
		int exp9k;

		branch = keyPNode + 1;
		B = (int) Math.ceil((double) size / (double) tuplesPLeaf);
		k = (int) Math.ceil(Math.log10((double) B)
				/ Math.log10((double) branch));
		exp9k = (int) Math.pow(branch, k);
		x = (exp9k - B) / keyPNode;

		nonLeaf = ((exp9k - 1) >> mbits) - x;
		nonLeaf = nonLeaf > 0 ? nonLeaf : 0;
		nInternal = nonLeaf;
		thres = ((exp9k - 1) >> mbits);
	}

	private void copyIndexKeyAtPos(byte[] dest, int dstIdx, byte[] src,
			int srcIdx) {
		System.arraycopy(src, srcIdx, dest, dstIdx, keySize);
	}

	/**
	 * Compare two keys from based n their types
	 * 
	 * @param key1 first key
	 * @param key2 second key
	 * @return 0 - if keys are equal +ve - if key1 > key2 -ve - if key1 < key2
	 */
	public float compareKeys(byte[] key1, byte[] key2) {
		switch (keyType) {
		case DataTypes.INT_TYPE:
			return (BinaryUtils.getInt(key1, 0) - BinaryUtils.getInt(key2, 0));
		case DataTypes.FLOAT_TYPE:
			return (BinaryUtils.getFloat(key1, 0) - BinaryUtils.getFloat(key2, 0));

		case DataTypes.CHAR_TYPE:
			return BinaryUtils
					.getString(key1, 0, keySize - OFFSET_BYTES)
					.compareTo(
							BinaryUtils
									.getString(key2, 0, keySize - OFFSET_BYTES));
		case DataTypes.DATE_TYPE:
			return BinaryUtils.getDate(key1, 0, keySize - OFFSET_BYTES)
					.compareTo(
							BinaryUtils.getDate(key2, 0, keySize - OFFSET_BYTES));
		default:
			return 0;
		}
	}

	public byte[] getIndexArray() {
		return internalNodes;
	}

	public long get(final String key) {
		switch (keyType) {
		case DataTypes.INT_TYPE:
			return get(BinaryUtils.toBytes(Integer.parseInt(key)));
		case DataTypes.FLOAT_TYPE:
			return get(BinaryUtils.toBytes(Float.parseFloat(key)));
		case DataTypes.CHAR_TYPE:
			return get(key.getBytes());
		case DataTypes.DATE_TYPE:
			return get(BinaryUtils.toBytes(key));
		default:
			return -1;
		}
	}

	public long get(final byte[] key) {
		int l, node;
		int p;

		long offset = 0;
		node = 0;
		while (node < nInternal) {
			p = node << mbits;
			l = binarySearch(internalNodes, key, p, p + keyPNode - 1);
			if (l < 0) {
				l = -(l + 1);
				if (l > 0)
					offset = BinaryUtils.getLong(internalNodes, (l - 1) * keySize
							+ keySize - OFFSET_BYTES);
				else
					offset = BinaryUtils.getLong(internalNodes, l * keySize
							+ keySize - OFFSET_BYTES);
			} else
				offset = BinaryUtils.getLong(internalNodes, l * keySize + keySize
						- OFFSET_BYTES);

			l -= p;
			node = node * (keyPNode + 1) + l + 1;
		}
		l = node - thres;
		l = (l << (mbits - 1)); // l is the entry index
		if (l < 0)
			l += size;

		// now at the leaves
		/*
		 * l = binarySearchData(content, key, l, l + (tuplesPLeaf - 1) < size ?
		 * l + (tuplesPLeaf - 1) : size - 1);
		 * 
		 * if (l >= 0) return content[(2 * l) + 1]; else return NOT_FOUND;
		 */

		// return the leaf offset directly
		// return l;
		return offset;
	}

	public int getLowKey(final byte[] data, final String key) {
		byte[] keyBytes = null;
		switch (keyType) {
		case DataTypes.INT_TYPE:
			keyBytes = BinaryUtils.toBytes(Integer.parseInt(key));
			break;
		case DataTypes.FLOAT_TYPE:
			keyBytes = BinaryUtils.toBytes(Float.parseFloat(key));
			break;
		case DataTypes.CHAR_TYPE:
			keyBytes = key.getBytes();
			break;
		case DataTypes.DATE_TYPE:
			keyBytes = BinaryUtils.toBytes(key);
			break;
		}
		if (keyBytes != null)
			return recLen
					* binarySearchLeafLeft(data, keyBytes, 0, tuplesPLeaf - 1);
		else
			return -1;
	}

	private int binarySearchLeafLeft(final byte[] array, final byte[] key,
			final int from, final int to) {
		int low = from;
		int high = to;

		for (; low <= high;) {
			int mid = (low + high) >> 1;
			byte[] midVal = new byte[keySize];
			copyIndexKeyAtPos(midVal, 0, array, mid * recLen + keyOffset);

			if (compareKeys(midVal, key) < 0)
				low = mid + 1;
			else if (compareKeys(midVal, key) > 0)
				high = mid - 1;
			else {
				// key found, ensure we got the leftmost one
				byte[] tmpKey = new byte[keySize];
				while (mid > 0) {
					copyIndexKeyAtPos(tmpKey, 0, array, (mid - 1) * recLen
							+ keyOffset);
					if (compareKeys(tmpKey, key) != 0)
						break;
					mid--;
				}
				return mid;
			}
		}
		return (low); // key not found.
	}

	public int getHighKey(final byte[] data, final String key) {
		int numOfRecords = data.length / recLen;
		byte[] keyBytes = null;
		switch (keyType) {
		case DataTypes.INT_TYPE:
			keyBytes = BinaryUtils.toBytes(Integer.parseInt(key));
			break;
		case DataTypes.FLOAT_TYPE:
			keyBytes = BinaryUtils.toBytes(Float.parseFloat(key));
			break;
		case DataTypes.CHAR_TYPE:
			keyBytes = key.getBytes();
			break;
		case DataTypes.DATE_TYPE:
			keyBytes = BinaryUtils.toBytes(key);
			break;
		}
		if (keyBytes != null)
			return recLen
					* binarySearchLeafRight(data, keyBytes, numOfRecords
							- tuplesPLeaf, numOfRecords - 1);
		else
			return -1;
	}

	private int binarySearchLeafRight(final byte[] array, final byte[] key,
			final int from, final int to) {
		int low = from;
		int high = to;

		for (; low <= high;) {
			int mid = (low + high) >> 1;
			byte[] midVal = new byte[keySize];
			copyIndexKeyAtPos(midVal, 0, array, mid * recLen + keyOffset);

			if (compareKeys(midVal, key) < 0)
				low = mid + 1;
			else if (compareKeys(midVal, key) > 0)
				high = mid - 1;
			else {
				// key found, ensure we got the rightmost one
				byte[] tmpKey = new byte[keySize];
				while (mid < high) {
					copyIndexKeyAtPos(tmpKey, 0, array, (mid + 1) * recLen
							+ keyOffset);
					if (compareKeys(tmpKey, key) != 0)
						break;
					mid++;
				}
				return mid + 1;
			}
		}
		return (low - 1); // key not found.
	}

	/**
	 * Binary search on the given array for given key in the range [from,to].
	 * 
	 * @param array
	 * @param key
	 * @param from
	 * @param to
	 * @return positive index of key in array, negative index of insert position
	 *         if key not found.
	 */
	private int binarySearch(final byte[] array, final byte[] key,
			final int from, final int to) {
		int low = from;
		int high = to;

		// int midOffset;

		for (; low <= high;) {
			int mid = (low + high) >> 1;
			byte[] midVal = new byte[keySize];
			copyIndexKeyAtPos(midVal, 0, array, mid * keySize);
			// midOffset = ByteUtils.getInt(midVal, keySize-OFFSET_BYTES);

			if (compareKeys(midVal, key) < 0)
				low = mid + 1;
			else if (compareKeys(midVal, key) > 0)
				high = mid - 1;
			else {
				// key found, ensure we got the leftmost one
				byte[] tmpKey = new byte[keySize];
				copyIndexKeyAtPos(tmpKey, 0, array, (mid - 1) * keySize);
				while (mid > 0 && compareKeys(tmpKey, key) == 0) {
					mid--;
					// midOffset = ByteUtils.getInt(tmpKey,
					// keySize-OFFSET_BYTES);
					copyIndexKeyAtPos(tmpKey, 0, array, (mid - 1) * keySize);
				}
				return mid;
				// return midOffset;
			}
		}

		return -(low + 1); // key not found.
	}

	/**
	 * Binary search on the given data array (keys and values) for given key in
	 * the range [from,to].
	 * 
	 * @param dataArray
	 * @param key
	 * @param from
	 * @param to
	 * @return positive index of key in data array, negative index of insert
	 *         position if key not found.
	 */
	/*
	 * private int binarySearchData(final int[] dataArray, final int key, final
	 * int from, final int to) { int low = from; int high = to;
	 * 
	 * for (; low <= high;) { int mid = (low + high) >> 1; long midVal =
	 * dataArray[2 * mid];
	 * 
	 * if (midVal < key) low = mid + 1; else if (midVal > key) high = mid - 1;
	 * else { // key found, ensure we got the leftmost one while (mid >= 2 &&
	 * dataArray[mid - 2] == key) { mid -= 2; } return mid; }
	 * 
	 * } return -(low + 1); // key not found. }
	 */

	public byte[] getHeader(long contentSize, byte[] firstKey, byte[] lastKey,
			int noOfRecords) {
		// Header contains: index offset(int), content offset(long), number of
		// tuples(int)
		// and two keys (first and last key in the index)
		byte[] header = new byte[16 + 2 * keySize];
		BinaryUtils.append(header, 0, internalNodes.length);
		BinaryUtils.append(header, 4, contentSize);
		BinaryUtils.append(header, 12, noOfRecords);
		System.arraycopy(firstKey, 0, header, 16, keySize);
		System.arraycopy(lastKey, 0, header, 16 + keySize, keySize);

		return header;
	}
}
